package org.example.flinksql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Print {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // todo:Connector 'print' can only be used as a sink. It cannot be used as a source.

        // create an input table
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_output (" +
                "id INT," +
                "name STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'print'"+
                ")");

        // get table
        Table t_output = tbEnv.from("t_output");

        /* print table struct
        root
                |-- id: INT
                |-- name: STRING
                |-- sex: STRING
                |-- age: INT
        */
        t_output.printSchema();

        tbEnv.executeSql("insert into t_output(id,name,sex,age) values(0,'name-0','女', 21)").print();


        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
